package com.ls.mq.actor.impl;

import com.google.common.base.Preconditions;
import com.ls.mq.iterface.ActorRef;
import com.ls.mq.iterface.IMQFuture;
import com.ls.mq.manager.FutureManager;
import com.ls.mq.message.Destination;
import com.ls.mq.message.Message;
import com.ls.mq.pubsub.PubSubProxy;
import com.ls.mq.queue.QueueProxy;
import com.ls.mq.util.Toolkit;
import com.sun.istack.internal.Nullable;

public class ActorRefImpl implements ActorRef {
	public static final int DEFAULTTIMEOUT = 3 * 1000;
	private Destination src;

	private Destination dest;
	private String uuid;
	private long timeout;

	public static ActorRefImpl build(Destination dest) {
		return new ActorRefImpl(dest);
	}

	public static ActorRefImpl build(Destination src,Destination dest) {
		return new ActorRefImpl(dest,src);
	}
	
	public static ActorRefImpl build(Message msg) {
		return new ActorRefImpl(msg.getDest(), msg.getSrc(), msg.getUuid());
	}

	/**
	 * 必须有dest，不能为空
	 * 
	 * @param src
	 */
	public ActorRefImpl(Destination dest) {
		this(dest, null);
	}

	/**
	 * 
	 * @param dest
	 *            目的地址不能为空
	 * @param src
	 */
	public ActorRefImpl(Destination dest, @Nullable Destination src) {
		this(dest, src, null);
	}

	public ActorRefImpl(Destination dest, @Nullable Destination src, String uuid) {
		this.dest = Preconditions.checkNotNull(dest);
		this.src = src;
		this.uuid = uuid;
	}

	@Override
	public void send(Object request) {
		Message msg = new Message(src, dest, request, true);
		msg.setRequest(true);
		msg.setRequestReply(false);
		QueueProxy.getInstance().publish(getDest().getChannel(), Toolkit.toJson(msg));
	}

	@Override
	public void sendDirectly(String msg) {
		QueueProxy.getInstance().publish(getDest().getChannel(), msg);

	}

	@Override
	public Object sendWithReply(Object request) {
		IMQFuture future = asyncSendReply(request);
		return future.get();
	}

	@Override
	public Object sendWithReply(Object request, int timeout) {
		IMQFuture future = asyncSendReply(request, timeout);
		return future.get();
	}

	@Override
	public IMQFuture asyncSendReply(Object request) {
		IMQFuture future = asyncSendReply(request, DEFAULTTIMEOUT);
		return future;
	}

	@Override
	public IMQFuture asyncSendReply(Object request, int timeout) {
		Message msg = new Message(src, dest, request, true);
		msg.setRequest(true);
		msg.setRequestReply(true);
		QueueProxy.getInstance().publish(getDest().getChannel(), Toolkit.toJson(msg));
		IMQFuture future = FutureManager.getInstance().addFuture(msg.getUuid(), timeout);
		return future;
	}

	@Override
	public void publish(Object request) {
		PubSubProxy.getInstance().publish(getDest().getChannel(), Toolkit.toJson(request));

	}

	@Override
	public void replyUnsafe(Object response) {
		Message msg = new Message(dest, src, response, true, getUuid());
		msg.setRequest(false);
		msg.setRequestReply(false);
		QueueProxy.getInstance().publish(getSrc().getChannel(), Toolkit.toJson(msg));

	}

	@Override
	public void setTimeout(int timeout) {
		// TODO Auto-generated method stub

	}

	public Destination getSrc() {
		return src;
	}

	public void setSrc(Destination src) {
		this.src = src;
	}

	public Destination getDest() {
		return dest;
	}

	public void setDest(Destination dest) {
		this.dest = dest;
	}

	public long getTimeout() {
		return timeout;
	}

	public void setTimeout(long timeout) {
		this.timeout = timeout;
	}

	public String getUuid() {
		return uuid;
	}

	public void setUuid(String uuid) {
		this.uuid = uuid;
	}

}
